package com.bbx.flink.dataStream_api_all.CDC;

import cn.hutool.json.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

/**
 * 自定义   debezium    反序列化
 */
public class CustomerDebeziumDeserializationSchema implements DebeziumDeserializationSchema<String> {
    private static final long serialVersionUID = 9042745421910624673L;


    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }

    /**
     * {
     * "db":"",
     * "table":"",
     * "before":{"id:"","name":""...},
     * "after":{"id:"","name":""...},
     * "op":""
     * }
     *
     * @param sourceRecord
     * @param collector
     * @throws Exception
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        JSONObject jsonObject = new JSONObject();
        //获取 db 和table
        String topic = sourceRecord.topic();
        String[] split = topic.split("\\.");
        //获取before,after
        Struct value = (Struct) sourceRecord.value();
        Object before = getStructInfo(value, "before");
        Object after = getStructInfo(value, "after");

        //获取操作类型
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);

        jsonObject.set("db", split[0])
                .set("table", split[1])
                .set("op", operation)
                .set("before", before)
                .set("after", after);
        collector.collect(jsonObject.toString());
    }



    public Object getStructInfo(Struct struct,String name){
        Struct tmpStruct = struct.getStruct(name);
        JSONObject jsonObject = new JSONObject();
        if (tmpStruct != null) {
            Schema schema = tmpStruct.schema();
            List<Field> fields = schema.fields();
            fields.forEach(i -> {
                jsonObject.set(i.name(), tmpStruct.get(i));
            });
        }
        return jsonObject;


    }
}
